#pragma once
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../comm/msg.pb.h"
#include "../comm/mq.pb.h"
#include "connection.h"
#include "virtualhost.h"
#include "consumer.h"

// 消息队列服务器是消息队列的核心组件，它负责管理消息队列，包括
// 存储消息、确保消息的顺序、维护消息的持久性、控制对消息队列的访问权限等诸多功能。
namespace ns_broker
{
    using namespace ns_conn;
    using namespace ns_virtual;
    using namespace ns_consume;
    using namespace placeholders;
    using namespace std;

    class BrokerServer
    {
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;

    public:
        BrokerServer(int port)
            : _svr(&_loop, muduo::net::InetAddress("0.0.0.0", port),
                   "Server", muduo::net::TcpServer::kReusePort),
              _dispatcher(std::bind(&BrokerServer::unkonwnmessage, this, _1, _2, _3)),
              _codec(make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, _1, _2, _3))),
              _virtual_host(make_shared<VirtualHost>("./data/message/", "./data/host1.db")),
              _consumer_manager(make_shared<ConsumerManage>()),
              _connection_manager(make_shared<ConnectionManage>()),
              _threadpool(make_shared<ThreadPool>())
        {
            // 队列在数据恢复时需要创建对应的消费者管理单元(unordered_map下的vector对象)
            unordered_map<string, Queue::ptr> allqueue = _virtual_host->allQueues();
            for (auto &qmap : allqueue)
            {
                _consumer_manager->init_queue_consumer(qmap.first);
            }

            // 注册_dispatcher业务请求处理函数
            _dispatcher.registerMessageCallback<openChannelRequest>(
                std::bind(&BrokerServer::createChannel, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<closeChannelRequest>(
                std::bind(&BrokerServer::closeChannel, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<declareExchangeRequest>(
                std::bind(&BrokerServer::declareExcahnge, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<deleteExchangeRequest>(
                std::bind(&BrokerServer::deleteExchange, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<declareQueueRequest>(
                std::bind(&BrokerServer::declareQueue, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<deleteQueueRequest>(
                std::bind(&BrokerServer::deleteQueue, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<queueBindRequest>(
                std::bind(&BrokerServer::queueBind, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<queueUnBindRequest>(
                std::bind(&BrokerServer::queueUnBind, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<basicPublishRequest>(
                std::bind(&BrokerServer::basicPublish, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<basicAckRequest>(
                std::bind(&BrokerServer::basicAck, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<basicConsumeRequest>(
                std::bind(&BrokerServer::basicConsume, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<basicCancelRequest>(
                std::bind(&BrokerServer::basicCancel, this, _1, _2, _3));

            // 连接回调处理
            _svr.setConnectionCallback(std::bind(&BrokerServer::myconnect, this, _1));

            // 设置成协议处理器，在协议处理结束后,其内部会再次调用请求分发器执行对应的回调函数
            _svr.setMessageCallback(std::bind(
                &ProtobufCodec::onMessage, _codec.get(), _1, _2, _3));
        }
        void start()
        {
            _svr.start(); // 创建套接字+将套接字设置到_loop(epoll)事件
            _loop.loop(); // 死循环阻塞,等待监听套接字事件到来,然后进行IO处理,回调_svr里设置的的回调函数
        }

    private:
        // 创建、关闭信道
        void createChannel(const muduo::net::TcpConnectionPtr &conn,
                           const openChannelRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+创建信道
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "创建信道失败，连接不存在" << endl;
                return;
            }
            myconn->createChannel(req);
        }
        void closeChannel(const muduo::net::TcpConnectionPtr &conn,
                          const closeChannelRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+关闭信道
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "关闭信道失败，连接不存在" << endl;
                return;
            }
            myconn->closeChannel(req);
        }

        // 声明、删除交换机
        void declareExcahnge(const muduo::net::TcpConnectionPtr &conn,
                             const declareExchangeRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "声明交换机失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "声明交换机失败，信道不存在" << endl;
                return;
            }
            mychannel->declare_exchange(req);
        }
        void deleteExchange(const muduo::net::TcpConnectionPtr &conn,
                            const deleteExchangeRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "删除交换机失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "删除交换机失败，信道不存在" << endl;
                return;
            }
            mychannel->delete_exchange(req);
        }

        // 声明、删除队列
        void declareQueue(const muduo::net::TcpConnectionPtr &conn,
                          const declareQueueRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "声明队列失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "声明队列失败，信道不存在" << endl;
                return;
            }
            mychannel->declare_queue(req);
        }
        void deleteQueue(const muduo::net::TcpConnectionPtr &conn,
                         const deleteQueueRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "删除队列失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "删除队列失败，信道不存在" << endl;
                return;
            }
            mychannel->delete_queue(req);
        }

        // 交换机队列绑定、解绑
        void queueBind(const muduo::net::TcpConnectionPtr &conn,
                       const queueBindRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "交换机队列绑定失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "交换机队列绑定失败，信道不存在" << endl;
                return;
            }
            mychannel->bind(req);
        }
        void queueUnBind(const muduo::net::TcpConnectionPtr &conn,
                         const queueUnBindRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "交换机队列解绑失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "交换机队列解绑失败，信道不存在" << endl;
                return;
            }
            mychannel->unbind(req);
        }

        // 消息发布-"生产者"
        void basicPublish(const muduo::net::TcpConnectionPtr &conn,
                          const basicPublishRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "消息发布失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "消息发布失败，信道不存在" << endl;
                return;
            }
            mychannel->basic_publish(req);
        }
        // 消息确认
        void basicAck(const muduo::net::TcpConnectionPtr &conn,
                      const basicAckRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "消息确认失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "消息确认失败，信道不存在" << endl;
                return;
            }
            mychannel->basic_ack(req);
        }

        // 订阅队列消息-"消费者客户端"
        void basicConsume(const muduo::net::TcpConnectionPtr &conn,
                          const basicConsumeRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "订阅队列消息失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "订阅队列消息失败，信道不存在" << endl;
                return;
            }
            mychannel->basic_consum(req);
        }
        // 取消订阅队列消息
        void basicCancel(const muduo::net::TcpConnectionPtr &conn,
                         const basicCancelRequestPtr &req, muduo::Timestamp)
        {
            // 获取连接+获取信道+调用信道提供的接口函数
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                LOG(WARNING) << "取消订阅队列消息失败，连接不存在" << endl;
                return;
            }
            Channel::ptr mychannel = myconn->getChannel(req->cid());
            if (mychannel.get() == nullptr)
            {
                LOG(WARNING) << "取消订阅队列消息失败，信道不存在" << endl;
                return;
            }
            mychannel->basic_cancel(req);
        }

        // ------------------------------------------------------------
        // 收到未知消息处理函数
        void unkonwnmessage(const muduo::net::TcpConnectionPtr &conn,
                            const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "未知信息: " << message->GetTypeName();
            conn->shutdown(); // 关闭连接
        }

        // 连接就绪(listen_socket)就绪的回调函数
        void myconnect(const muduo::net::TcpConnectionPtr &conn)
        {
            // 连接建立成功立即在内存中实例化一个连接对象
            if (conn->connected())
            {
                _connection_manager->createConnection(
                    _consumer_manager, conn, _codec, _virtual_host, _threadpool);
                cout << "新连接建立成功" << endl;
            }
            else
            {
                _connection_manager->closeConnection(conn);
                cout << "连接关闭" << endl;
            }
        }

    private: //(_svr内部调用_codec,_codec内部调用_dispatcher)
        // EventLoop对象必须在TcpServer对象之前，因为TcpServer对象在初始化列表中用到了EventLoop
        muduo::net::EventLoop _loop;    // epoll事件监听+IO操作
        muduo::net::TcpServer _svr;     // 设置回调处理数据请求
        ProtobufDispatcher _dispatcher; // 请求分发器-注册请求处理函数
        ProtobufCodecPtr _codec;        // 协议处理器-将收到的请求数据进行协议处理

        shared_ptr<VirtualHost> _virtual_host;            // 虚拟机
        shared_ptr<ConsumerManage> _consumer_manager;     // 消费者
        shared_ptr<ConnectionManage> _connection_manager; // 连接
        // 通过连接找到信道，然后通过信道进行一系列的操作
        shared_ptr<ThreadPool> _threadpool; // 线程池
    };
}